package com.zy.gmall.realtime.util;

import com.zy.gmall.realtime.common.GmallConfig;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;

import java.util.Properties;

public class DorisUtil {
    public static DorisSink<String> getDorisSink(String table){

        Properties properties = new Properties();
        properties.setProperty("format","json");
        properties.setProperty("read_json_by_line","true");

        return DorisSink.<String>builder()
                .setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisOptions(DorisOptions.builder()
                        .setFenodes(GmallConfig.DORIS_FE)
                        .setTableIdentifier(GmallConfig.DORIS_DB + "." + table)
                        .setUsername("root")
                        .setPassword("000000")
                        .build())
                .setDorisExecutionOptions(DorisExecutionOptions.builder()
                        //.setLabelPrefix("doris-label") //streamload label prefix
                        .disable2PC() // 开启两阶段提交后,labelPrefix 需要全局唯一,为了测试方便禁用两阶段提交
                        .setDeletable(false)
                        .setBufferCount(3) // 批次条数: 默认 3
                        .setBufferSize(8 * 1024) // 批次大小: 默认 1M
                        .setCheckInterval(3000) // 批次输出间隔   三个对批次的限制是或的关系
                        .setMaxRetries(3)
                        .setStreamLoadProp(properties) // 设置 stream load 的数据格式 默认是 csv,需要改成 json
                        .build())
                .setSerializer(new SimpleStringSerializer())
                .build() ;
    }
}
